<!DOCTYPE html>
<html prefix="og: http://ogp.me/ns# article: http://ogp.me/ns/article# " lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>4.Using the threading and concurrent.futures Modules | 绿萝间</title>
<link href="../assets/css/all-nocdn.css" rel="stylesheet" type="text/css">
<link href="../assets/css/ipython.min.css" rel="stylesheet" type="text/css">
<link href="../assets/css/nikola_ipython.css" rel="stylesheet" type="text/css">
<meta name="theme-color" content="#5670d4">
<meta name="generator" content="Nikola (getnikola.com)">
<link rel="alternate" type="application/rss+xml" title="RSS" href="../rss.xml">
<link rel="canonical" href="https://muxuezi.github.io/posts/4using-the-threading-and-concurrentfutures-modules.html">
<script type="text/x-mathjax-config">
MathJax.Hub.Config({
    tex2jax: {
        inlineMath: [ ['$','$'], ["\\(","\\)"] ],
        displayMath: [ ['$$','$$'], ["\\[","\\]"] ],
        processEscapes: true
    },
    displayAlign: 'center', // Change this to 'center' to center equations.
    "HTML-CSS": {
        styles: {'.MathJax_Display': {"margin": 0}}
    }
});
</script><!--[if lt IE 9]><script src="../assets/js/html5.js"></script><![endif]--><meta name="author" content="Tao Junjie">
<link rel="prev" href="slott-2015113-2015-09-27-report.html" title="双色球2015113期(2015-09-27)数据分析报告" type="text/html">
<link rel="next" href="dlott-15113-2015-09-28-report.html" title="大乐透15113期(2015-09-28)数据分析报告" type="text/html">
<meta property="og:site_name" content="绿萝间">
<meta property="og:title" content="4.Using the threading and concurrent.futures Modules">
<meta property="og:url" content="https://muxuezi.github.io/posts/4using-the-threading-and-concurrentfutures-modules.html">
<meta property="og:description" content="4.用threading和concurrent.futures模块¶在上一章，我们总结了并行思想可以解决的一些问题。这一章，我们将用Python的threading模块实现每个问题的解决方案。
本章内容包括以下主题：

线程的定义
threading和_thread的选择
用threading实现多请求的Fibonacci数列
用concurrent.futures模块实现网络爬虫







">
<meta property="og:type" content="article">
<meta property="article:published_time" content="2015-09-28T20:42:03+08:00">
<meta property="article:tag" content="Parallel Programming with Python">
<meta property="article:tag" content="Python">
</head>
<body>
<a href="#content" class="sr-only sr-only-focusable">Skip to main content</a>

<!-- Menubar -->

<nav class="navbar navbar-inverse navbar-static-top"><div class="container">
<!-- This keeps the margins nice -->
        <div class="navbar-header">
            <button type="button" class="navbar-toggle collapsed" data-toggle="collapse" data-target="#bs-navbar" aria-controls="bs-navbar" aria-expanded="false">
            <span class="sr-only">Toggle navigation</span>
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
            </button>
            <a class="navbar-brand" href="https://muxuezi.github.io/">

                <span id="blog-title">绿萝间</span>
            </a>
        </div>
<!-- /.navbar-header -->
        <div class="collapse navbar-collapse" id="bs-navbar" aria-expanded="false">
            <ul class="nav navbar-nav">
<li>
<a href="../archive.html">Archive</a>
                </li>
<li>
<a href="../categories/">Tags</a>
                </li>
<li>
<a href="../rss.xml">RSS feed</a>

                
            </li>
</ul>
<ul class="nav navbar-nav navbar-right">
<li>
    <a href="4using-the-threading-and-concurrentfutures-modules.ipynb" id="sourcelink">Source</a>
    </li>

                
            </ul>
</div>
<!-- /.navbar-collapse -->
    </div>
<!-- /.container -->
</nav><!-- End of Menubar --><div class="container" id="content" role="main">
    <div class="body-content">
        <!--Body content-->
        <div class="row">
            
            
<article class="post-text h-entry hentry postpage" itemscope="itemscope" itemtype="http://schema.org/Article"><header><h1 class="p-name entry-title" itemprop="headline name"><a href="#" class="u-url">4.Using the threading and concurrent.futures Modules</a></h1>

        <div class="metadata">
            <p class="byline author vcard"><span class="byline-name fn">
                    Tao Junjie
            </span></p>
            <p class="dateline"><a href="#" rel="bookmark"><time class="published dt-published" datetime="2015-09-28T20:42:03+08:00" itemprop="datePublished" title="2015-09-28 20:42">2015-09-28 20:42</time></a></p>
            
        <p class="sourceline"><a href="4using-the-threading-and-concurrentfutures-modules.ipynb" id="sourcelink">Source</a></p>

        </div>
        

    </header><div class="e-content entry-content" itemprop="articleBody text">
    <div tabindex="-1" id="notebook" class="border-box-sizing">
    <div class="container" id="notebook-container">

<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<h2 id="4.用threading和concurrent.futures模块">4.用<code>threading</code>和<code>concurrent.futures</code>模块<a class="anchor-link" href="4using-the-threading-and-concurrentfutures-modules.html#4.%E7%94%A8threading%E5%92%8Cconcurrent.futures%E6%A8%A1%E5%9D%97">¶</a>
</h2>
<p>在上一章，我们总结了并行思想可以解决的一些问题。这一章，我们将用Python的<code>threading</code>模块实现每个问题的解决方案。</p>
<p>本章内容包括以下主题：</p>
<ul>
<li>线程的定义</li>
<li>
<code>threading</code>和<code>_thread</code>的选择</li>
<li>用<code>threading</code>实现多请求的Fibonacci数列</li>
<li>用<code>concurrent.futures</code>模块实现网络爬虫</li>
</ul>
<!-- TEASER_END-->
</div>
</div>
</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<h3 id="线程的定义">线程的定义<a class="anchor-link" href="4using-the-threading-and-concurrentfutures-modules.html#%E7%BA%BF%E7%A8%8B%E7%9A%84%E5%AE%9A%E4%B9%89">¶</a>
</h3>
<p>线程（thread）是一个进程（process）中不同执行单元（execution line）。如果把一个程序比喻成蜂巢（hive），那么向蜂巢里收集花粉（pollen）这件事就是一个进程。这个收集花粉的进程是由若干只为了解决花粉不足问题的工蜂（worker bee）同时工作来完成。这些工蜂扮演的角色就是线程，在进程内部执行并共享资源以完成各自的任务。</p>
<p>线程属于同一个进程并共享同一块内存。因此，开发者的任务就是控制和使用这块内存。</p>
<h4 id="使用线程的优缺点">使用线程的优缺点<a class="anchor-link" href="4using-the-threading-and-concurrentfutures-modules.html#%E4%BD%BF%E7%94%A8%E7%BA%BF%E7%A8%8B%E7%9A%84%E4%BC%98%E7%BC%BA%E7%82%B9">¶</a>
</h4>
<p>使用线程之前，需要考虑线程具有的一些优缺点，而且线程与具体的编程语言和操作系统有密切关联。</p>
<p>线程的优点主要有：</p>
<ul>
<li>线程在同一个进程内的通信速度，数据定位速度，和信息共享速度都是非常快的</li>
<li>创建线程的成本比创建进程低，因为它不需要复制主进程上下文所包含的全部信息</li>
<li>通过处理器缓存优化内存接入速度可以高效使用数据局部存储（data locality，thread local storage，TLS）</li>
</ul>
<p>线程的缺点主要有：</p>
<ul>
<li>数据共享允许快速通信。但是，有时会让经验不足的开发者犯错。</li>
<li>数据共享限制了解决方案的灵活性。比如，迁移到分布性架构时，线程是十分令人头疼的事情。通常，线程会限制算法的扩展性。</li>
</ul>
<blockquote>
<p>在Python程序语言里，由于GIL的存在，使用带有CPU限制的线程可能影响程序的性能。</p>
</blockquote>
<h4 id="线程类型">线程类型<a class="anchor-link" href="4using-the-threading-and-concurrentfutures-modules.html#%E7%BA%BF%E7%A8%8B%E7%B1%BB%E5%9E%8B">¶</a>
</h4>
<p>用两种线程：内核线程（kernel thread）与用户线程（user thread）。内核线程是由操作系统创建和管理的线程。线程上下文的切换（Context Switch），调度，和结束都是通过操作系统的内核管理的。而用户线程的所有状态是由开发者控制的。</p>
<p>我们可以对比两者的优缺点。</p>
<p>内核线程的优点主要有：</p>
<ul>
<li>一个内核线程常只轻量进程（lightweight processes）。因此如果一个内核线程阻塞，其他内核线程还可以运行</li>
<li>内核线程可以在不同的CPU上运行</li>
</ul>
<p>内核线程的缺点主要有：</p>
<ul>
<li>创建与同步操作开销都较大</li>
<li>实现方式与平台有关</li>
</ul>
<p>用户线程的优点主要有：</p>
<ul>
<li>创建与同步操作开销都较小</li>
<li>实现方式与平台无关</li>
</ul>
<p>用户线程的缺点主要有：</p>
<ul>
<li>一个进程内的所有用户线程只能与一个内核线程关联。因此，如果一个用户线程阻塞，其他用户线程都不能执行</li>
<li>用户线程不运行在不同的CPU上</li>
</ul>
<h4 id="线程状态定义">线程状态定义<a class="anchor-link" href="4using-the-threading-and-concurrentfutures-modules.html#%E7%BA%BF%E7%A8%8B%E7%8A%B6%E6%80%81%E5%AE%9A%E4%B9%89">¶</a>
</h4>
<p>线程生命周期中一共五个状态，如下所示：</p>
<ul>
<li>
<strong>创建</strong>：主进程创建一个线程，创建之后，会被发送到线程队列准备执行</li>
<li>
<strong>执行</strong>：这个阶段，线程使用CPU</li>
<li>
<strong>就绪</strong>：这个阶段，线程在线程队列里准备就绪即将执行</li>
<li>
<strong>阻塞</strong>：这个阶段，线程被阻塞，可能是等待I/O操作完成，这个阶段不使用CPU</li>
<li>
<strong>结束</strong>：这个阶段，释放执行时占用的资源，结束线程的生命周期</li>
</ul>
<h4 id="threading和_thread的选择">
<code>threading</code>和<code>_thread</code>的选择<a class="anchor-link" href="4using-the-threading-and-concurrentfutures-modules.html#threading%E5%92%8C_thread%E7%9A%84%E9%80%89%E6%8B%A9">¶</a>
</h4>
<p>Python标准库里提供了两个线程模块：<code>_thread</code>模块（这个Python模块提供了一个线程的底层API，具体文档见<em><code>https://docs.python.org/3.5/library/_thread.html#module-_thread</code></em>）和<code>threading</code>模块（这个Python模块提供了一个线程的高层API，具体文档见<em><code>https://docs.python.org/3.5/library/threading.html#module-threading</code></em>）。这个<code>threading</code>模块通过了比<code>_thread</code>模块更友好的接口，使用起来更方便。开发者根据自己口味选择。如果开发者觉得用线程的底层API更舒服，可以实现自己的线程池（thread pool），互斥锁和其他原始特征，那么他可以选择<code>_thread</code>模块。否则，还是用<code>threading</code>模块最方便。</p>

</div>
</div>
</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<h3 id="用threading实现多请求的Fibonacci数列">用<code>threading</code>实现多请求的Fibonacci数列<a class="anchor-link" href="4using-the-threading-and-concurrentfutures-modules.html#%E7%94%A8threading%E5%AE%9E%E7%8E%B0%E5%A4%9A%E8%AF%B7%E6%B1%82%E7%9A%84Fibonacci%E6%95%B0%E5%88%97">¶</a>
</h3>
<p>现在是时候回到现实。我们的任务是在Fibonacci数列有多个输入请求时并行的执行。处于教学的目的，我们把输入值设置成四个，每个输入值分配一个线程运行函数，让worker与要执行的任务（task）完美地一一对应。算法如下所示：</p>
<ol>
<li>首先，用列表储存需要计算的四个值，然后把值发送到一个允许线程同步使用的结构中。</li>
<li>当等待计算的值被发送到可同步使用的结构中之后，计算Fibonacci数列的线程需要一个信号告诉它们数据已经准备就绪了。我们用一个叫<code>Condition</code>的线程同步机制（Thread synchronization mechanism）。（<code>Condition</code>机制是一个Python对象，用来提供多个线程之间数据使用的同步机制，具体文档查看<em><code>https://docs.python.org/3/library/threading.html#condition-objects</code></em>。）</li>
<li>当每个线程完成对应Fibonacci数列的计算之后，把结果保存到词典中。</li>
</ol>
<p>那么，现在让我们演示最有趣的代码和注释内容。</p>
<p>代码开始的时候，我们使用Unicode编码格式，然后导入<code>logging</code>，<code>threading</code>和<code>Queue</code>模块。另外，我们在例子中已经定义了主要的数据结构。我们定义了一个字典<code>fibo_dict</code>，把每个整数（作为用户输入）存储为键，对应Fibonacci数列的计算作为字典的值。我们还用<code>queue</code>模块中的<code>Queue</code>模块声明了一个<code>Queue</code>对象<code>shared_queue</code>，用来存储计算Fibonacci数列线程之间的共享数据，线程把数据插入<code>Queue</code>对象。最后，我们定义最后一个数据结构——一个带四个元素的Python的<code>list</code>对象，模拟程序即将接收到的值。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="c1">#coding: utf-8</span>
<span class="kn">import</span> <span class="nn">logging</span><span class="o">,</span> <span class="nn">threading</span>

<span class="kn">from</span> <span class="nn">queue</span> <span class="k">import</span> <span class="n">Queue</span>

<span class="n">logger</span> <span class="o">=</span> <span class="n">logging</span><span class="o">.</span><span class="n">getLogger</span><span class="p">()</span>
<span class="n">logger</span><span class="o">.</span><span class="n">setLevel</span><span class="p">(</span><span class="n">logging</span><span class="o">.</span><span class="n">DEBUG</span><span class="p">)</span>
<span class="n">formatter</span> <span class="o">=</span> <span class="n">logging</span><span class="o">.</span><span class="n">Formatter</span><span class="p">(</span><span class="s1">'</span><span class="si">%(asctime)s</span><span class="s1"> - </span><span class="si">%(message)s</span><span class="s1">'</span><span class="p">)</span>

<span class="n">ch</span> <span class="o">=</span> <span class="n">logging</span><span class="o">.</span><span class="n">StreamHandler</span><span class="p">()</span>
<span class="n">ch</span><span class="o">.</span><span class="n">setLevel</span><span class="p">(</span><span class="n">logging</span><span class="o">.</span><span class="n">DEBUG</span><span class="p">)</span>
<span class="n">ch</span><span class="o">.</span><span class="n">setFormatter</span><span class="p">(</span><span class="n">formatter</span><span class="p">)</span>
<span class="n">logger</span><span class="o">.</span><span class="n">addHandler</span><span class="p">(</span><span class="n">ch</span><span class="p">)</span>

<span class="n">fibo_dict</span> <span class="o">=</span> <span class="p">{}</span>
<span class="n">shared_queue</span> <span class="o">=</span> <span class="n">Queue</span><span class="p">()</span>
<span class="n">input_list</span> <span class="o">=</span> <span class="p">[</span><span class="mi">3</span><span class="p">,</span> <span class="mi">10</span><span class="p">,</span> <span class="mi">5</span><span class="p">,</span> <span class="mi">7</span><span class="p">]</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<blockquote>
<p><strong>下载源代码</strong></p>
<p>你可以在Packt网站<em><code>https://www.packtpub.com/</code></em>，用你的账号下载所有书籍的示例代码。如果你从其他地方买的书，你可以先用<em><code>https://www.packtpub.com/books/content/support</code></em>搜索书籍名称，然后用你的邮箱地址注册一个账号下载代码。</p>
</blockquote>
<p>下面一行代码，我们将定义一个<code>threading</code>模块里的<code>Condition</code>对象。这个对象可以通过一个具体的条件来同步资源使用的状态。</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="n">queue_condition</span> <span class="o">=</span> <span class="n">threading</span><span class="o">.</span><span class="n">Condition</span><span class="p">()</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>使用<code>Condition</code>对象是为了控制队列的创建，以及里面发生的过程。</p>
<p>代码的另一个部分是定义线程即将执行的函数<code>fibonacci_task</code>。<code>fibonacci_task</code>函数的参数是<code>condition</code>对象，用来控制<code>fibonacci_task</code>使用<code>shared_queue</code>。我们使用<code>with</code>语句（关于<code>with</code>语句更多的信息，请参考<em><code>https://docs.python.org/3/reference/compound_stmts.html#the-with-statement</code></em>）来简化上下文的形式。如果不使用<code>with</code>语句，我们就需要显式地对线程锁进行获取和释放。使用<code>with</code>语句，我们就可以在程序一开始就获取线程锁，然后在程序结束的时候自动退出线程锁。在<code>fibonacci_task</code>函数的<code>with</code>语句后面是一个逻辑表达式，告诉当前的线程，“当<code>shared_queue</code>为空时，一直等待”。等待就是用<code>condition</code>对象的<code>wait()</code>方法是实现的。线程会等待，直到它被告知<code>shared_queue</code>可以使用时才停止。一旦条件得到满足，就会立即计算Fibonacci数列的值并存储到字典<code>fibo_dict</code>中。最后，我们调用<code>task_done()</code>方法，确认某一个排队的任务已经被抽取出来执行了。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="k">def</span> <span class="nf">fibonacci_task</span><span class="p">(</span><span class="n">condition</span><span class="p">):</span>
    <span class="k">with</span> <span class="n">condition</span><span class="p">:</span>
        <span class="k">while</span> <span class="n">shared_queue</span><span class="o">.</span><span class="n">empty</span><span class="p">():</span>
            <span class="n">logger</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="s2">"[</span><span class="si">%s</span><span class="s2">] - waiting for elements in queue.."</span>
                <span class="o">%</span> <span class="n">threading</span><span class="o">.</span><span class="n">current_thread</span><span class="p">()</span><span class="o">.</span><span class="n">name</span><span class="p">)</span>
            <span class="n">condition</span><span class="o">.</span><span class="n">wait</span><span class="p">()</span>
        <span class="k">else</span><span class="p">:</span>
            <span class="n">value</span> <span class="o">=</span> <span class="n">shared_queue</span><span class="o">.</span><span class="n">get</span><span class="p">()</span>
            <span class="n">a</span><span class="p">,</span> <span class="n">b</span> <span class="o">=</span> <span class="mi">0</span><span class="p">,</span> <span class="mi">1</span>
            <span class="k">for</span> <span class="n">item</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="n">value</span><span class="p">):</span>
                <span class="n">a</span><span class="p">,</span> <span class="n">b</span> <span class="o">=</span> <span class="n">b</span><span class="p">,</span> <span class="n">a</span> <span class="o">+</span> <span class="n">b</span>
                <span class="n">fibo_dict</span><span class="p">[</span><span class="n">value</span><span class="p">]</span> <span class="o">=</span> <span class="n">a</span>
        <span class="n">shared_queue</span><span class="o">.</span><span class="n">task_done</span><span class="p">()</span>
        <span class="n">logger</span><span class="o">.</span><span class="n">debug</span><span class="p">(</span><span class="s2">"[</span><span class="si">%s</span><span class="s2">] fibonacci of key [</span><span class="si">%d</span><span class="s2">] with result [</span><span class="si">%d</span><span class="s2">]"</span> <span class="o">%</span>
            <span class="p">(</span><span class="n">threading</span><span class="o">.</span><span class="n">current_thread</span><span class="p">()</span><span class="o">.</span><span class="n">name</span><span class="p">,</span> <span class="n">value</span><span class="p">,</span> <span class="n">fibo_dict</span><span class="p">[</span><span class="n">value</span><span class="p">]))</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>我们定义的第二个函数是<code>queue_task</code>是由线程执行的，把线程计算的结果添加到<code>shared_queue</code>里。我们会发现函数把<code>condition</code>作为线程同步接入<code>shared_queue</code>队列的参数。线程会把<code>input_list</code>中的每个元素都存储到<code>shared_queue</code>队列中。</p>
<p>当元素都存储到<code>shared_queue</code>队列之后，函数会告诉线程负责计算Fibonacci数列的队列已经准备好了。这个函数用<code>condition.notifyAll()</code>完成，如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="k">def</span> <span class="nf">queue_task</span><span class="p">(</span><span class="n">condition</span><span class="p">):</span>
    <span class="n">logging</span><span class="o">.</span><span class="n">debug</span><span class="p">(</span><span class="s1">'Starting queue_task...'</span><span class="p">)</span>
    <span class="k">with</span> <span class="n">condition</span><span class="p">:</span>
        <span class="k">for</span> <span class="n">item</span> <span class="ow">in</span> <span class="n">input_list</span><span class="p">:</span>
            <span class="n">shared_queue</span><span class="o">.</span><span class="n">put</span><span class="p">(</span><span class="n">item</span><span class="p">)</span>
        <span class="n">logging</span><span class="o">.</span><span class="n">debug</span><span class="p">(</span><span class="s2">"Notifying fibonacci_task threads that the queue is ready to consume.."</span><span class="p">)</span>
        <span class="n">condition</span><span class="o">.</span><span class="n">notifyAll</span><span class="p">()</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>在后面的代码里，我们创建一个<code>shared_queue</code>准备条件已经满足的四线程组。然后我们使用一个允许自定义函数的<code>thread</code>类构造器（constructor），这个线程将用<code>target</code>参数设置要执行函数，在<code>args</code>里面放入要执行函数的参数，如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="n">threads</span> <span class="o">=</span> <span class="p">[</span><span class="n">threading</span><span class="o">.</span><span class="n">Thread</span><span class="p">(</span><span class="n">daemon</span><span class="o">=</span><span class="kc">True</span><span class="p">,</span> <span class="n">target</span><span class="o">=</span><span class="n">fibonacci_task</span><span class="p">,</span>
                            <span class="n">args</span><span class="o">=</span><span class="p">(</span><span class="n">queue_condition</span><span class="p">,))</span> <span class="k">for</span> <span class="n">i</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="mi">4</span><span class="p">)]</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>然后，我们开启准备计算Fibonacci数列的线程。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="p">[</span><span class="n">thread</span><span class="o">.</span><span class="n">start</span><span class="p">()</span> <span class="k">for</span> <span class="n">thread</span> <span class="ow">in</span> <span class="n">threads</span><span class="p">]</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>紧接着，我们再创建并执行一个控制<code>shared_queue</code>的线程。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="n">prod</span> <span class="o">=</span> <span class="n">threading</span><span class="o">.</span><span class="n">Thread</span><span class="p">(</span><span class="n">name</span><span class="o">=</span><span class="s1">'queue_task_thread'</span><span class="p">,</span> <span class="n">daemon</span><span class="o">=</span><span class="kc">True</span><span class="p">,</span> 
                        <span class="n">target</span><span class="o">=</span><span class="n">queue_task</span><span class="p">,</span> <span class="n">args</span><span class="o">=</span><span class="p">(</span><span class="n">queue_condition</span><span class="p">,))</span>
<span class="n">prod</span><span class="o">.</span><span class="n">start</span><span class="p">()</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>最后，我们对所有线程调用<code>join()</code>方法来计算Fibonacci数列。这么做的目的是让<code>main</code>线程等待这些线程全部执行完Fibonacci数列，这样它就不会在其他线程还没有结束之前就停止。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="p">[</span><span class="n">thread</span><span class="o">.</span><span class="n">join</span><span class="p">()</span> <span class="k">for</span> <span class="n">thread</span> <span class="ow">in</span> <span class="n">threads</span><span class="p">]</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>程序的执行结果如下图所示：
<img src="http://muxuezi.github.io/posts/ppp/ch4/threading_fibonacci.png" alt="">
可以看出，一开始<code>fibonacci_task</code>线程被创建并初始化，紧接着进入等待状态。同时，<code>queue_task</code>被创建，然后生成<code>shared_queue</code>。最后，<code>queue_task</code>告诉<code>fibonacci_task</code>线程可以执行任务了。</p>
<p>还会看到，<code>fibonacci_task</code>线程执行并不是自然顺序，每次运行线程执行顺序都会变化。这就是线程的特点：不确定性。</p>
<p><a href="http://muxuezi.github.io/posts/ppp/ch4/parallel_fibonacci.py">源代码</a></p>

</div>
</div>
</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<h3 id="用concurrent.futures模块并行抓取网页">用<code>concurrent.futures</code>模块并行抓取网页<a class="anchor-link" href="4using-the-threading-and-concurrentfutures-modules.html#%E7%94%A8concurrent.futures%E6%A8%A1%E5%9D%97%E5%B9%B6%E8%A1%8C%E6%8A%93%E5%8F%96%E7%BD%91%E9%A1%B5">¶</a>
</h3>
<p>下面我们将用代码来实现一个并行网络爬虫。这里我们会用一个非常有意思的Python功能，<code>concurrent.futures</code>模块中的<code>ThreadPoolExecutor</code>。在前面的例子中，我们用非常原始的线程功能实现了<code>parallel_fibonacci.py</code>。而且我们是通过手工操作创建和初始化多个线程。在线程较多的项目中，这些做很难管理。通常，都会使用线程池解决线程的状态管理问题。线程池是一种用来保存若干线程的结构，在进程里使用之前创建。其目的是重用线程，这样可以避免重复创建线程——降低资源消耗。</p>
<p>下面的内容基本和上一节一样，我们需要一个通过多个阶段来执行不同任务的算法，这些任务是彼此相关的。我们将演示并行网络爬虫的代码。</p>
<p>导入模块并设置日志文件之后，我们用Python的<a href="https://docs.python.org/3/library/re.html#module-re"><code>re</code>模块</a>新建一个正则表达式，我们将在抓取阶段用这个表达式从网页源代码中筛选出所有链接。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="n">html_link_regex</span> <span class="o">=</span> <span class="n">re</span><span class="o">.</span><span class="n">compile</span><span class="p">(</span><span class="s1">'&lt;a\s(?:.*?\s)*?href=[</span><span class="se">\'</span><span class="s1">"](.*?)[</span><span class="se">\'</span><span class="s1">"].*?&gt;'</span><span class="p">)</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>紧接着，我们用一个同步队列来模拟多个输入URL任务。然后，新建一个字典<code>result_dict</code>来存储结果，把URL和对应页面的链接列表存在字典中。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="n">urls</span> <span class="o">=</span> <span class="n">queue</span><span class="o">.</span><span class="n">Queue</span><span class="p">()</span>
<span class="n">urls</span><span class="o">.</span><span class="n">put</span><span class="p">(</span><span class="s1">'http://www.sina.com'</span><span class="p">)</span>
<span class="n">urls</span><span class="o">.</span><span class="n">put</span><span class="p">(</span><span class="s1">'http://cn.bing.com/'</span><span class="p">)</span>
<span class="n">urls</span><span class="o">.</span><span class="n">put</span><span class="p">(</span><span class="s1">'https://coding.net/'</span><span class="p">)</span>
<span class="n">urls</span><span class="o">.</span><span class="n">put</span><span class="p">(</span><span class="s1">'http://github.com/'</span><span class="p">)</span>
<span class="n">urls</span><span class="o">.</span><span class="n">put</span><span class="p">(</span><span class="s1">'http://mail.126.com/'</span><span class="p">)</span>

<span class="n">result_dict</span> <span class="o">=</span> <span class="p">{}</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>之后我们定义一个函数<code>group_urls_task</code>，从同步队列中抽取URL加入到字典<code>result_dict</code>里，URL作为字典<code>result_dict</code>的键。还有一个细节要注意是同步队列的<code>get()</code>函数用了两个参数。一个参数是<code>True</code>用来阻塞同步队列接入。另一个参数是<code>0.05</code>秒，避免在同步队列中没有任务时等待太长时间。有时候，你不会愿意花太多时间阻塞线程来等待任务出现。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="k">def</span> <span class="nf">group_urls_task</span><span class="p">(</span><span class="n">urls</span><span class="p">):</span>
    <span class="k">try</span><span class="p">:</span>
        <span class="n">url</span> <span class="o">=</span> <span class="n">urls</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="kc">True</span><span class="p">,</span> <span class="mf">0.05</span><span class="p">)</span>
        <span class="n">result_dict</span><span class="p">[</span><span class="n">url</span><span class="p">]</span> <span class="o">=</span> <span class="kc">None</span>
        <span class="n">logger</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="s2">"[</span><span class="si">%s</span><span class="s2">] putting url [</span><span class="si">%s</span><span class="s2">] in dictionary..."</span> <span class="o">%</span> <span class="p">(</span>
            <span class="n">threading</span><span class="o">.</span><span class="n">current_thread</span><span class="p">()</span><span class="o">.</span><span class="n">name</span><span class="p">,</span> <span class="n">url</span><span class="p">))</span>
    <span class="k">except</span> <span class="n">queue</span><span class="o">.</span><span class="n">Empty</span><span class="p">:</span>
        <span class="n">logging</span><span class="o">.</span><span class="n">error</span><span class="p">(</span><span class="s1">'Nothing to be done, queue is empty'</span><span class="p">)</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>现在，我们用函数<code>crawl_task</code>实现每个URL的页面的抓取。基本上抓取阶段就是获取URL对应页面上的所有链接。函数返回值是一个元组，第一个元素是URL，第二个元素是URL对应页面上的所有链接构成的列表。这里我们用<a href="http://docs.python-requests.org/en/latest/"><code>requests</code>模块</a>获取URL对应页面的内容。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="k">def</span> <span class="nf">crawl_task</span><span class="p">(</span><span class="n">url</span><span class="p">):</span>
    <span class="n">links</span> <span class="o">=</span> <span class="p">[]</span>
    <span class="k">try</span><span class="p">:</span>
        <span class="n">request_data</span> <span class="o">=</span> <span class="n">requests</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="n">url</span><span class="p">)</span>
        <span class="n">logger</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="s2">"[</span><span class="si">%s</span><span class="s2">] crawling url [</span><span class="si">%s</span><span class="s2">] ..."</span> <span class="o">%</span> <span class="p">(</span>
            <span class="n">threading</span><span class="o">.</span><span class="n">current_thread</span><span class="p">()</span><span class="o">.</span><span class="n">name</span><span class="p">,</span> <span class="n">url</span><span class="p">))</span>
        <span class="n">links</span> <span class="o">=</span> <span class="n">html_link_regex</span><span class="o">.</span><span class="n">findall</span><span class="p">(</span><span class="n">request_data</span><span class="o">.</span><span class="n">text</span><span class="p">)</span>
    <span class="k">except</span><span class="p">:</span>
        <span class="n">logger</span><span class="o">.</span><span class="n">error</span><span class="p">(</span><span class="n">sys</span><span class="o">.</span><span class="n">exc_info</span><span class="p">()[</span><span class="mi">0</span><span class="p">])</span>
        <span class="k">raise</span>
    <span class="k">finally</span><span class="p">:</span>
        <span class="k">return</span> <span class="p">(</span><span class="n">url</span><span class="p">,</span> <span class="n">links</span><span class="p">)</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>进一步分析代码，我们会看到里面创建了一个<code>concurrent.futures</code>模块中的<a href="https://docs.python.org/3.5/library/concurrent.futures.html"><code>ThreadPoolExecutor</code>对象</a>。在<code>ThreadPoolExecutor</code>对象的构造器里，设置了<code>max_workers</code>参数。这个参数在线程池里定义了供执行器（executor）使用的线程数量。在前面将URL从同步队列移除并作为键增加到字典<code>result_dict</code>这个阶段，都是用这个三个worker线程完成的。这个数量可以根据问题的需要进行设置。完成<code>ThreadPoolExecutor</code>对象的定义之后，我们用<code>with</code>语句保证线程的结束例程（ending routine），这些例程将会在<code>with</code>语句范围结束后立即执行。在<code>ThreadPoolExecutor</code>对象的范围内，我们在同步队列中重复它，以<code>submit</code>方法对同步队列中的URL执行一个引用。总之，<code>submit</code>方法为执行安排了一个可调用函数，然后返回一个带有创建执行的例程安排（scheduling）的<code>Future</code>对象。<code>submit</code>方法接收一个可调用函数和函数的参数；在我们的例子里，可调用函数就是<code>group_urls_task</code>，函数的参数就是同步队列<code>urls</code>。当这些参数被调用之后，线程池中的worker线程会以一种并行，异步的方式执行函数预订（booking）。代码如下所示：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="k">with</span> <span class="n">concurrent</span><span class="o">.</span><span class="n">futures</span><span class="o">.</span><span class="n">ThreadPoolExecutor</span><span class="p">(</span><span class="n">max_workers</span><span class="o">=</span><span class="mi">3</span><span class="p">)</span> <span class="k">as</span> <span class="n">group_link_threads</span><span class="p">:</span>
    <span class="k">for</span> <span class="n">i</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="n">urls</span><span class="o">.</span><span class="n">qsize</span><span class="p">()):</span>
        <span class="n">group_link_threads</span><span class="o">.</span><span class="n">submit</span><span class="p">(</span><span class="n">group_urls_task</span><span class="p">,</span> <span class="n">urls</span><span class="p">)</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>之后我们又定义了一个新的<code>ThreadPoolExecutor</code>对象；但是这一次我们想用前面<code>group_urls_task</code>函数生成的字典<code>result_dict</code>的键来执行抓取阶段的任务。代码会有些不同：</p>

</div>
</div>
</div>
<div class="cell border-box-sizing code_cell rendered">
<div class="input">
<div class="prompt input_prompt">In [ ]:</div>
<div class="inner_cell">
    <div class="input_area">
<div class=" highlight hl-ipython3"><pre><span></span><span class="k">with</span> <span class="n">concurrent</span><span class="o">.</span><span class="n">futures</span><span class="o">.</span><span class="n">ThreadPoolExecutor</span><span class="p">(</span><span class="n">max_workers</span><span class="o">=</span><span class="mi">3</span><span class="p">)</span> <span class="k">as</span> <span class="n">crawler_link_threads</span><span class="p">:</span>
    <span class="n">future_tasks</span> <span class="o">=</span> <span class="p">{</span><span class="n">crawler_link_threads</span><span class="o">.</span><span class="n">submit</span><span class="p">(</span><span class="n">crawl_task</span><span class="p">,</span> <span class="n">url</span><span class="p">):</span> <span class="n">url</span> <span class="k">for</span> <span class="n">url</span> <span class="ow">in</span> <span class="n">result_dict</span><span class="o">.</span><span class="n">keys</span><span class="p">()}</span>
    <span class="k">for</span> <span class="n">future</span> <span class="ow">in</span> <span class="n">concurrent</span><span class="o">.</span><span class="n">futures</span><span class="o">.</span><span class="n">as_completed</span><span class="p">(</span><span class="n">future_tasks</span><span class="p">):</span>
        <span class="n">result_dict</span><span class="p">[</span><span class="n">future</span><span class="o">.</span><span class="n">result</span><span class="p">()[</span><span class="mi">0</span><span class="p">]]</span> <span class="o">=</span> <span class="n">future</span><span class="o">.</span><span class="n">result</span><span class="p">()[</span><span class="mi">1</span><span class="p">]</span>
</pre></div>

</div>
</div>
</div>

</div>
<div class="cell border-box-sizing text_cell rendered">
<div class="prompt input_prompt">
</div>
<div class="inner_cell">
<div class="text_cell_render border-box-sizing rendered_html">
<p>我们映射了一个临时字典<code>future_tasks</code>。里面包含了<code>submit</code>方法进行的函数预订，参数是<code>result_dict</code>的每个URL键。也就是说，我们用<code>future_tasks</code>为每个键创建了一个入口（entry）。映射之后，我们需通过<code>concurrent.futures.as_completed(fs, timeout=None)</code>方法来寻找<code>future_tasks</code>里入口，通过一个循环来获取前面函数预订的结果。这个调用会返回一个由<code>Future</code>对象实例组成的迭代器（iterator）。因此，我们可以迭代出每一个已经被处理好的预订结果。在<code>ThreadPoolExecutor</code>对象的最后，我们用<code>Future</code>对象<code>result()</code>方法获取抓取线程的结果。在这个例子的抓取阶段，<code>Future</code>对象返回的结果是元组。用这种方式我们获得的<code>future_tasks</code>最终结果如下图所示：
<img src="http://muxuezi.github.io/posts/ppp/ch4/threading_webcrawler.png" alt="">
我们再一次看到线程池中的线程出现的顺序并不是自然顺序，这是由于线程具有不确定性。重要的是通过打印<code>result_dict</code>的项目来展示最终结果。</p>
<p><a href="http://muxuezi.github.io/posts/ppp/ch4/parallel_web_crawler.py">源代码</a></p>
<h3 id="本章小结">本章小结<a class="anchor-link" href="4using-the-threading-and-concurrentfutures-modules.html#%E6%9C%AC%E7%AB%A0%E5%B0%8F%E7%BB%93">¶</a>
</h3>
<p>在这一章里，我们重点介绍了线程的概念。我们通过<code>threading</code>和<code>concurrent.futures</code>模块实现了上一章里提到的两个问题。通过问题的解决展示了两个模块的原理和灵活性。</p>
<p>下一章，我们将重点介绍如何用<code>multiprocessing</code>和<code>ProcessPoolExecutor</code>解决这两个问题。</p>

</div>
</div>
</div>
    </div>
  </div>

    </div>
    <aside class="postpromonav"><nav><ul itemprop="keywords" class="tags">
<li><a class="tag p-category" href="../categories/parallel-programming-with-python.html" rel="tag">Parallel Programming with Python</a></li>
            <li><a class="tag p-category" href="../categories/python.html" rel="tag">Python</a></li>
        </ul>
<ul class="pager hidden-print">
<li class="previous">
                <a href="slott-2015113-2015-09-27-report.html" rel="prev" title="双色球2015113期(2015-09-27)数据分析报告">Previous post</a>
            </li>
            <li class="next">
                <a href="dlott-15113-2015-09-28-report.html" rel="next" title="大乐透15113期(2015-09-28)数据分析报告">Next post</a>
            </li>
        </ul></nav></aside><script type="text/javascript" src="https://cdn.mathjax.org/mathjax/latest/MathJax.js?config=TeX-AMS-MML_HTMLorMML"> </script><script type="text/x-mathjax-config">
MathJax.Hub.Config({
    tex2jax: {
        inlineMath: [ ['$','$'], ["\\(","\\)"] ],
        displayMath: [ ['$$','$$'], ["\\[","\\]"] ],
        processEscapes: true
    },
    displayAlign: 'center', // Change this to 'center' to center equations.
    "HTML-CSS": {
        styles: {'.MathJax_Display': {"margin": 0}}
    }
});
</script></article>
</div>
        <!--End of body content-->

        <footer id="footer">
            Contents © 2017         <a href="mailto:muxuezi@gmail.com">Tao Junjie</a> - Powered by         <a href="https://getnikola.com" rel="nofollow">Nikola</a>         
<a rel="license" href="http://creativecommons.org/licenses/by-nc-sa/4.0">
<img alt="Creative Commons License BY-NC-SA" style="border-width:0; margin-bottom:12px;" src="http://i.creativecommons.org/l/by-nc-sa/4.0/80x15.png"></a>
            
        </footer>
</div>
</div>


            <script src="../assets/js/all-nocdn.js"></script><script>$('a.image-reference:not(.islink) img:not(.islink)').parent().colorbox({rel:"gal",maxWidth:"100%",maxHeight:"100%",scalePhotos:true});</script><!-- fancy dates --><script>
    moment.locale("en");
    fancydates(0, "YYYY-MM-DD HH:mm");
    </script><!-- end fancy dates --><script>
  (function(i,s,o,g,r,a,m){i['GoogleAnalyticsObject']=r;i[r]=i[r]||function(){
  (i[r].q=i[r].q||[]).push(arguments)},i[r].l=1*new Date();a=s.createElement(o),
  m=s.getElementsByTagName(o)[0];a.async=1;a.src=g;m.parentNode.insertBefore(a,m)
  })(window,document,'script','//www.google-analytics.com/analytics.js','ga');

  ga('create', 'UA-51330059-1', 'auto');
  ga('send', 'pageview');

</script>
</body>
</html>
